package cn.edu.flink.scala.tutorial.topn

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable

object ProcessAllWindowFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val stream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp).map(_.url)

    stream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
      .process(new ProcessAllWindowFunction[String, String, TimeWindow] {
        override def process(context: Context, elements: Iterable[String], out: Collector[String]): Unit = {
          val map = mutable.Map[String, Long]()

          elements.foreach(url => {
            map.get(url) match {
              case Some(count) => map.put(url, count + 1)
              case None => map.put(url, 1L)
            }
          })

          val result = map.toList.sortBy(-_._2).take(3)

          val builder = new StringBuilder()

          builder.append(s"============窗口：${context.window.getStart} ~ ${context.window.getEnd}============\n")
          for (i <- result.indices){
            builder.append(s"Top${i+1} url: ${result(i)._1}, count: ${result(i)._2} \n")
          }

          out.collect(builder.toString())
        }
      })
      .print()

    env.execute("TopNWithAllWindow")
  }
}
